package com.atguigu.flink.state;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * Created by Smexy on 2022/12/17
 *
 *      适用于动态更新配置的场景。
 *
 *              处理数据流： 数据流
 *              规则(属性)配置的流: 配置流， 做成 广播流
 *
 *              数据流在处理数据时，读取配置流中的信息。配置流中的信息可能需要动态变化，希望配置流中的信息一旦变化，可以广播到所有的数据流的Task中，
 *              让数据流用最新更新后的配置再进行处理。
 */
public class Demo2_BroadCastState
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //获取数据流
        SingleOutputStreamOperator<WaterSensor> dataDS = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction());

        //获取配置流
        SingleOutputStreamOperator<MyConf> configDS = env.socketTextStream("hadoop103", 8889)
                                                          .map(new MapFunction<String, MyConf>()
                                                          {
                                                              @Override
                                                              public MyConf map(String value) throws Exception {
                                                                  String[] words = value.split(",");
                                                                  return new MyConf(words[0], words[1]);
                                                              }
                                                          });

        MapStateDescriptor<String, String> mapStateDescriptor = new MapStateDescriptor<>("conf", String.class, String.class);
        //制作配置流为广播流
        BroadcastStream<MyConf> broadcastStream = configDS.broadcast(mapStateDescriptor);

        dataDS.connect(broadcastStream)
              .process(new BroadcastProcessFunction<WaterSensor, MyConf, WaterSensor>()
              {
                  //处理数据流
                  @Override
                  public void processElement(WaterSensor value, ReadOnlyContext ctx, Collector<WaterSensor> out) throws Exception {
                      ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

                      String newValue = broadcastState.get(value.getId());

                      value.setId(newValue);

                      out.collect(value);

                  }

                  //处理广播流
                  @Override
                  public void processBroadcastElement(MyConf value, Context ctx, Collector<WaterSensor> out) throws Exception {

                      //获取广播状态
                      BroadcastState<String, String> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

                      //把当前广播流中新流入的数据存入状态。自动地广播到所有处理数据流的Task
                      broadcastState.put(value.getName(),value.getValue());
                  }
              })
              .print();


        try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    private static class MyConf{
        private String name;
        private String value;
    }
}
